package com.jerry.frame.rpc.registry;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by zjh on 2017/8/9.
 */
public class ZookeeperRegistryService extends AbstractRegistryService {

    private CuratorFramework configClient;
    //会话超时时间
    private final int sessionTimeoutMs = 60 * 1000;
    //连接超时时间
    private final int connetionTimeoutMs = 15 * 1000;
    //启动服务提供地址，默认是127.0.0.1
    private final String address = "127.0.0.1";
    //存放被订阅zk上服务节点
    private final ConcurrentMap<RegistryMeta.ServiceMeta,PathChildrenCache> pathChildrenCaches = new ConcurrentHashMap<RegistryMeta.ServiceMeta, PathChildrenCache>();

    public void connectRegistryServer(String registryAddresses) {
        //500重连间隔时间，20 重连次数
        //创建访问zookeeper的客户端
        configClient = CuratorFrameworkFactory.newClient(registryAddresses, sessionTimeoutMs, connetionTimeoutMs, new ExponentialBackoffRetry(500, 20));
        //增加重新连接监听，这边是指在之前连接的时候已经注册或者订阅一部分服务，重连的时候需要重新把这些信息发布到zk上
        configClient.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                //重新订阅
                for (RegistryMeta.ServiceMeta serviceMeta : subscribeSet) {
                    doSubscribe(serviceMeta);

                }
                //重新发布服务
                for (RegistryMeta meta : registryMetaMap.keySet()) {
                    ZookeeperRegistryService.super.register(meta);

                }
            }
        });
        configClient.start();
    }

    public void subscribe(RegistryMeta.ServiceMeta serviceMeta) {
        String directory = String.format()
      PathChildrenCache pathChildrenCache = pathChildrenCaches.get(serviceMeta);
        if(pathChildrenCache==null){
            PathChildrenCache newPathChildrenCache = new PathChildrenCache(configClient,,false);
        }
    }

    protected void doUnregister(RegistryMeta registryMeta) {
        String directory = String.format("/jupiter/provider/%s/%s/%s",
                registryMeta.getServiceMeta().getGroup(),
                registryMeta.getServiceMeta().getServiceProviderName(),
                registryMeta.getServiceMeta().getVersion());
        try {
            if(configClient.checkExists().forPath(directory)==null){
                return;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        registryMeta.getAddress().setHost(address);

        try {
            configClient.delete().inBackground(new BackgroundCallback() {

                public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {

                }
            }).forPath(
                    String.format("%s/%s:%s:%s:%s",
                            directory,
                            registryMeta.getAddress().getHost(),
                            String.valueOf(registryMeta.getAddress().getPort()),
                            String.valueOf(registryMeta.getWeight()),
                            String.valueOf(registryMeta.getConnCount())));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void doSubscribe(RegistryMeta.ServiceMeta serviceMeta) {

    }

    public void destroy() {

    }
//注册服务
    public void doRegisterService(final RegistryMeta registryMeta) throws Exception {
        RegistryMeta.ServiceMeta serviceMeta = registryMeta.getServiceMeta();
        //拼接zk路径
        String directory = String.format("jerry/rpc/provider/%s/%s/%s", serviceMeta.getGroup(), serviceMeta.getServiceProviderName(), serviceMeta.getVersion());
        //创建服务路径，如果不存的话

        if (configClient.checkExists().forPath(directory) == null) {
            configClient.create().creatingParentsIfNeeded().forPath(directory);
        }
        registryMeta.getAddress().setHost(address);
        /*
        * CreateMode 有四种
        * 1.PERSISTENT 持久化，即使断开连接也不会消失
        * 2.PERSISTENT_SEQUENTIAL 持久化，且会在节点后追加序列号
        * 3.EPHEMERAL 临时节点，断开连接会消失
        * 4.EPHEMERAL_SEQUENTIAL 临时序列化节点，断开即消失
        * */

        //使用临时节点
        configClient.create().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            //在节点创建完回调，将创建成功的服务设置为完成
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                if (curatorEvent.getResultCode() == KeeperException.Code.OK.intValue()) {
                    registryMetaMap.put(registryMeta, RegistryState.DONE);
                }
            }
        }).forPath( String.format("%s/%s:%s:%s:%s",
                directory,
                registryMeta.getAddress().getHost(),
                String.valueOf(registryMeta.getAddress().getPort()),
                String.valueOf(registryMeta.getWeight()),
                String.valueOf(registryMeta.getConnCount())));

    }

    public void doCheckRegisterNodeStatus() {

    }
}
